package org.ala.tiktools.rpc.plugin.load_balance;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.client_server.client.AlaRpcClient;
import org.ala.tiktools.rpc.config.RpcConfigInfo;
import org.ala.tiktools.rpc.plugin.load_balance.pool.NodePool;
import org.ala.tiktools.rpc.plugin.load_balance.selector.ISelector;
import org.ala.tiktools.rpc.plugin.node.INode;
import org.ala.tiktools.rpc.plugin.node.INodeCreator;
import org.ala.tiktools.rpc.plugin.node.NodeKey;
import org.ala.tiktools.rpc.plugin.node.NodeStatusEnums;
import org.ala.tiktools.tools.list.Lists;
import org.slf4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;


/**
 * LoadBalance 负载均衡器
 *
 * @author ala
 * @date 2025-04-07 10:12
 */
public abstract class AbstractLoadBalance implements ILoadBalance {


    static Logger log = LoggerFactory.getLogger("rpc", "loadBalance");


    /**
     *  全部节点池
     */
    protected NodePool pool = new NodePool();
    /**
     *  节点key和group关系
     */
    protected Map<NodeKey, Set<String>> reverseGroups = new ConcurrentHashMap<>();
    /**
     *  活跃的node
     */
    protected Map<String, CopyOnWriteArrayList<INode>> activeNodes = new ConcurrentHashMap<>();
    /**
     *  定时任务
     *  <p> 全局唯一就好，用不了那么多
     */
    public static ScheduledExecutorService schedule = new ScheduledThreadPoolExecutor(2);


    /**
     * 节点构建器
     */
    @SuppressWarnings("rawtypes")
	protected INodeCreator nodeCreator;
    /**
     *  选择器
     */
    protected ISelector selector;
    /**
     *
     */
    protected ApplicationContext context;
    /**
     *
     */
    protected Environment env;


    /**
     * 初始化方法
     * <p>  对于私有的loadBalance，每个代理都会new一个新的，所以要自行初始化
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        context = applicationContext;
    }
    @Override
    public void setEnvironment(Environment environment) {
        env = environment;
    }


    @Override
    public INode select(String group) {
        //  如果loadBalance暂时取不到，等待500ms后继续检测。但总共不超过一个消息超时单位
    	//	有可能所有服务整体重启了，让请求飞一会
        long start = System.currentTimeMillis();
        CopyOnWriteArrayList<INode> nodes = activeNodes.get(group);
        while (Lists.isEmpty(nodes) && (System.currentTimeMillis() - start) < RpcConfigInfo.INSTANCE.getTimeout()) {
            try {
                Thread.sleep(500);
                nodes = activeNodes.get(group);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        if (nodes == null || nodes.isEmpty()) { return null; }
        return selector.select(nodes);
    }

    @Override
    public void loadRpcClient(AlaRpcClient anno) {
        //  如果group不为空，从group中初始化节点
        if (!StringUtils.isEmpty(anno.name())) {
            loadRpcNodeKeyByGroup(anno.name(), k -> appendNode(anno.name(), k));
        } else if (!StringUtils.isEmpty(anno.hosts())) {
            parseHosts(anno.hosts(), k -> appendNode(anno.hosts(), k));
        }
    }
    /**
     *  追加节点
     */
    protected void appendNode(String group, NodeKey k) {
        INode node = pool.get(k);
        if (node == null) {
            INode _node = nodeCreator.createNode(group, k);
            //  挂载节点失效监听
            _node.addStatusListener(NodeStatusEnums.UNAVAILABLE, () -> {
                //  从活跃列表中摘除
                removeFromActives(group, _node);
                //  重试连接
                log.info("[{} removeFromActives] ({}ms)后尝试重连 node:{}", this.getClass().getSimpleName(), RpcConfigInfo.INSTANCE.getReActiveInterval(), _node.key());
                schedule.schedule(_node::active, RpcConfigInfo.INSTANCE.getReActiveInterval(), TimeUnit.MILLISECONDS);
            });
            pool.appendAndActive(_node);
            node = _node;
        }
        //  绑定反向group，如果绑定成功，挂载节点有效监听
        if (reverseGroups.computeIfAbsent(k, _k -> new HashSet<>()).add(group)) {
            INode _node = node;
            node.addStatusListener(NodeStatusEnums.ACTIVATE, () -> appendToActives(group, _node));
        }
    }
    /**
     *  摘除节点
     */
    protected void removeNode(NodeKey k) {
        INode node = pool.remove(k);
        if (node != null) {
            //  从可用节点中删除
            Set<String> groups = reverseGroups.remove(k);
            if (groups != null) {
                groups.forEach(g -> removeFromActives(g, node));
            }
            //  自杀
            node.kill();
        }
    }
    /**
     *  把node放入活跃列表中
     */
    protected void appendToActives(String group, INode node) {
        //  把node放入活跃队列中
        CopyOnWriteArrayList<INode> nodes = activeNodes.get(group);
        if (nodes == null) {
        	synchronized (this) {
        		nodes = activeNodes.get(group);
        		if (nodes == null) {
        			schedule.schedule(RePopNodeTask.build().group(group).loadBalance(this), RpcConfigInfo.INSTANCE.getRePopNodeInterval(), TimeUnit.MILLISECONDS);
                    nodes = new CopyOnWriteArrayList<>();
                    activeNodes.put(group, nodes);
        		}
			}
        }
        //  检测node是否在nodes中
        for (INode n : nodes) {
            if (n.equals(node)) {
                log.info("[{} appendToActives] 节点已经在活跃队列中，可供select group:{} node:{}", this.getClass().getSimpleName(), group, node.key());
                return;
            }
        }
        //	需要确保add操作是串行的
        nodes.add(node);
        log.info("[{} appendToActives] 节点进入活跃队列，可供select id:{} group:{} node:{} activeList.size:{}", this.getClass().getSimpleName(), System.identityHashCode(this), group, node.key(), nodes.size());
    }
    /**
     *  把node从活跃列表清除
     */
    protected void removeFromActives(String group, INode node) {
        CopyOnWriteArrayList<INode> nodes = activeNodes.get(group);
        if (nodes != null) {
            //  检测node是否在nodes中
            int idx = -1;
            for (int i = 0; i < nodes.size(); i++) {
                if (nodes.get(i).equals(node)) {
                    idx = i;
                    break;
                }
            }
            if (idx != -1) {
                nodes.remove(idx);
                log.info("[{} removeFromActives] 节点从活跃队列删除 group:{} node:{}", this.getClass().getSimpleName(), group, node.key());
            }
        }
    }
    /**
     *  从group取节点信息
     */
    protected abstract void loadRpcNodeKeyByGroup(String group, Consumer<NodeKey> consumer);
    /**
     *  初始化节点
     */
    protected INode createNode(String group, NodeKey nodeKey) {
        return nodeCreator.createNode(group, nodeKey);
    }
    /**
     *  按照{host:port;host2:port2}格式解析hosts
     */
    protected void parseHosts(String hosts, Consumer<NodeKey> consumer) {
    	String[] ss = hosts.split(";");
    	for (String _ss : ss) {
    		String[] sss = _ss.split(":");
    		String host = sss[0];
    		int[] ports = parsePorts(sss[1]);
    		for (int port : ports) {
    			consumer.accept(NodeKey.build(host, port));
    		}
    	}
//        char[] cs = hosts.toCharArray();
//        int l = 0;
//        String host = null; int port;
//        for (int i = 0 ; i < cs.length ; i++) {
//            if (cs[i] == ' ') {continue;}
//            if (cs[i] == ':') {
//                host = new String(cs, l, i-l);
//                l = i+1;
//            } else if (cs[i] == ';') {
//                port = 0;
//                for (int j = i-1, b = 1 ; j >= l ; j--, b *= 10) {
//                    port += (cs[j] - '0') * b;
//                }
//                consumer.accept(NodeKey.build().host(host).ports(new int[] {port}));
//                l = i+1;
//            }
//        }
//        if (l < cs.length) {
//            port = 0;
//            for (int j = cs.length-1, b = 1 ; j >= l ; j--, b *= 10) {
//                port += (cs[j] - '0') * b;
//            }
//            consumer.accept(NodeKey.build().host(host).ports(new int[] {port}));
//        }
    }
    protected int[] parsePorts(String sports) {
    	String[] ss = sports.split(",");
    	return Arrays.stream(ss).map(s -> Integer.parseInt(s)).mapToInt(Integer::new).toArray();
    }


    /**
     *  上报延迟
     */
    @Override
    public void rtt(NodeKey k, long rtt) {
        selector.rtt(k, rtt);
    }


    @SuppressWarnings("rawtypes")
	public INodeCreator getNodeCreator() {
        return nodeCreator;
    }

    @SuppressWarnings("rawtypes")
	public void setNodeCreator(INodeCreator nodeCreator) {
        this.nodeCreator = nodeCreator;
    }

    public ISelector getSelector() {
        return selector;
    }

    public void setSelector(ISelector selector) {
        this.selector = selector;
    }

    /**
     *  每间隔一段时间重新拉取节点信息任务
     */
    static class RePopNodeTask implements Runnable {
        protected String group;
        protected AbstractLoadBalance loadBalance;
        @Override
        public void run() {
            try {
                loadBalance.loadRpcNodeKeyByGroup(group, k -> {
                    loadBalance.appendNode(group, k);
                });
            } catch (Exception e) {
                log.warn(String.format("[AbstractLoadBalance RePopNodeTask.run] 重新拉取节点失败 group:%s", group), e);
            } finally {
                schedule.schedule(this, RpcConfigInfo.INSTANCE.getRePopNodeInterval(), TimeUnit.MILLISECONDS);
            }
        }
        public static RePopNodeTask build() {return new RePopNodeTask();}
        public String getGroup() {
            return group;
        }
        public void setGroup(String group) {
            this.group = group;
        }
        public RePopNodeTask group(String group) {
            setGroup(group);
            return this;
        }
        public AbstractLoadBalance getLoadBalance() {
            return loadBalance;
        }
        public void setLoadBalance(AbstractLoadBalance loadBalance) {
            this.loadBalance = loadBalance;
        }
        public RePopNodeTask loadBalance(AbstractLoadBalance loadBalance) {
            setLoadBalance(loadBalance);
            return this;
        }
    }
}
